package cn.wangjie.spark.sink.mysql

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.sql.{ForeachWriter, Row}

class MySQLForeachWriter extends ForeachWriter[Row] {

  // 定义变量
  var conn: Connection = _
  var pstmt: PreparedStatement = _
	val insertSQL = "REPLACE INTO `tb_word_count` (`id`, `word`, `count`) VALUES (NULL, ?, ?)"
	
	// 打开连接，比如连接数据连接
	override def open(partitionId: Long, epochId: Long): Boolean = {
		// a. 加载驱动类
		Class.forName("com.mysql.cj.jdbc.Driver")
		// b. 获取连接
		conn = DriverManager.getConnection(
			"jdbc:mysql://node1.itcast.cn:3306/db_spark?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true", //
			"root",
			"123456"
		)
		// c. 获取PreparedStatement
		pstmt = conn.prepareStatement(insertSQL)
		
		// 直接返回true，表示获取连接成功
		true
	}
	
	// 插入数据，比如将数据插入到MySQL表汇总
	override def process(row: Row): Unit = {
		// 设置参数
		pstmt.setString(1, row.getAs[String]("value"))
		pstmt.setLong(2, row.getAs[Long]("count"))
		// 执行插入
		pstmt.executeUpdate()
	}
	
	// 关闭连接
	override def close(errorOrNull: Throwable): Unit = {
		if(null != pstmt) pstmt.close()
		if(null != conn) conn.close()
	}
	
}
